package tman.simulator.core;

import common.configuration.Configuration;
import common.configuration.CyclonConfiguration;
import common.configuration.Global;
import common.configuration.MonitorConfiguration;
import common.configuration.TManConfiguration;
import common.events.CollectPeerData;
import common.events.PeerDataRequest;
import common.events.PeerDataResponse;
import common.simulation.SimulatorPort;
import common.simulation.GenerateReport;
import common.simulation.PeerFail;
import common.simulation.ConsistentHashtable;
import common.simulation.PeerJoin;
import java.math.BigInteger;
import java.util.HashMap;
import java.util.TreeMap;

import common.peer.JoinTMan;
import common.peer.JoinPeer;
import common.peer.PeerAddress;
import common.peer.PeerPort;
import common.simulation.SimulatorInit;

import se.sics.kompics.ChannelFilter;
import se.sics.kompics.Component;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.Start;
import se.sics.kompics.Stop;
import se.sics.kompics.address.Address;
import se.sics.kompics.network.Message;
import se.sics.kompics.network.Network;
import se.sics.kompics.p2p.bootstrap.BootstrapConfiguration;
import se.sics.kompics.p2p.overlay.OverlayAddress;
import se.sics.kompics.timer.SchedulePeriodicTimeout;
import se.sics.kompics.timer.Timer;
import se.sics.kompics.web.Web;
import se.sics.kompics.web.WebRequest;

import java.net.InetAddress;
import java.util.Random;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.sics.ipasdistances.AsIpGenerator;
import tman.system.peer.Peer;
import tman.system.peer.PeerInit;
import tman.system.peer.tman.PeerData;
import tman.simulator.snapshot.Snapshot;
import tman.simulator.snapshot.StateUtil;

public final class TManSimulator extends ComponentDefinition {
	Positive<SimulatorPort> simulator = positive(SimulatorPort.class);
	Positive<Network> network = positive(Network.class);
	Positive<Timer> timer = positive(Timer.class);
	Negative<Web> web = negative(Web.class);

	private static final Logger logger = LoggerFactory
			.getLogger(TManSimulator.class);
	
	private final HashMap<BigInteger, Component> peers;
	private final HashMap<BigInteger, PeerAddress> peersAddress;
	
	private Address peer0Address;
	private BootstrapConfiguration bootstrapConfiguration;
	private CyclonConfiguration cyclonConfiguration;
	private TManConfiguration tmanConfiguration;
	private MonitorConfiguration monitorConfiguration;

	private int peerIdSequence;
	private BigInteger cyclonIdentifierSpaceSize;
	private ConsistentHashtable<BigInteger> cyclonView;
    private AsIpGenerator ipGenerator;
    private Random rnd;
    private Integer reportThreshold;
    private Integer totalPeers;

//-------------------------------------------------------------------	
	public TManSimulator() {
		reportThreshold = 20;
		peers = new HashMap<BigInteger, Component>();
		peersAddress = new HashMap<BigInteger, PeerAddress>();
		cyclonView = new ConsistentHashtable<BigInteger>();
		totalPeers = 0;
		
		for(Integer OD: Global.odDistr.keySet())
		{
			totalPeers += Global.odDistr.get(OD);
		}
		
		totalPeers++;

                rnd = new Random(Global.seed);
                ipGenerator = AsIpGenerator.getInstance(rnd.nextInt());
                
		subscribe(handleInit, control);
		// UNCOMMENT TO ENABLE REPORT GENERATION USING THE SNAPSHOT CLASS
		//subscribe(handleGenerateReport, timer);
		subscribe(handlePeerJoin, simulator);
		subscribe(handlePeerFail, simulator);
		subscribe(handleCollectPeerData, simulator);
	}

//-------------------------------------------------------------------	
	Handler<SimulatorInit> handleInit = new Handler<SimulatorInit>() {
		public void handle(SimulatorInit init) {
			peers.clear();
			peerIdSequence = 0;

			peer0Address = init.getPeer0Address();
			bootstrapConfiguration = init.getBootstrapConfiguration();
			cyclonConfiguration = init.getCyclonConfiguration();
			tmanConfiguration = init.getTmanConfiguration();
			monitorConfiguration = init.getMonitorConfiguration();

			cyclonIdentifierSpaceSize = cyclonConfiguration.getIdentifierSpaceSize();
			
			// generate periodic report
			int snapshotPeriod = Configuration.SNAPSHOT_PERIOD;
			SchedulePeriodicTimeout spt = new SchedulePeriodicTimeout(snapshotPeriod, snapshotPeriod);
			spt.setTimeoutEvent(new GenerateReport(spt));
			trigger(spt, timer);
		}
	};

//-------------------------------------------------------------------	
	Handler<PeerJoin> handlePeerJoin = new Handler<PeerJoin>() {
		public void handle(PeerJoin event) {
			//System.out.println("Joining new peer!");
			int num = event.getNum();
			boolean isSource = event.isSource();
			BigInteger id = event.getPeerId();
			//System.out.println("Initial ID: " + id + " BW: " + num);
			// join with the next id if this id is taken
			BigInteger successor = cyclonView.getNode(id);

			while (successor != null && successor.equals(id)) {
				id = id.add(BigInteger.ONE).mod(cyclonIdentifierSpaceSize);
				successor = cyclonView.getNode(id);
			}

			Component newPeer = createAndStartNewPeer(id, num, isSource);
			cyclonView.addNode(id);
			if(isSource)
				System.out.println("Source Node ID: " + id);

			trigger(new JoinPeer(id), newPeer.getPositive(PeerPort.class));
			float percentageJoined = peersAddress.keySet().size()/(float)totalPeers*100;
//			System.out.println("PJ: " + percentageJoined);
//			System.out.println("RT: " + reportThreshold);
			if(percentageJoined>=reportThreshold)
			{
				System.out.println(reportThreshold + "% of peers have joined");
				reportThreshold += 20;
			}
			
			if(percentageJoined== 100.0)
			{
				for(BigInteger id2: peers.keySet())
				{
					trigger(new JoinTMan(), peers.get(id2).getPositive(PeerPort.class));
				}
			}
		}
	};
	
//-------------------------------------------------------------------	
	Handler<PeerFail> handlePeerFail = new Handler<PeerFail>() {
		public void handle(PeerFail event) {
			if(!event.isRemoveBatch())
			{
				BigInteger id = cyclonView.getNode(event.getCyclonId());
	
				if (cyclonView.size() == 0) {
					System.err.println("Empty network");
					return;
				}
	
				cyclonView.removeNode(id);
				stopAndDestroyPeer(id);
			}
			else
			{
				Random r = new Random(peers.keySet().size());
				for(int i=0; i<event.getCyclonId().intValue(); i++)
				{
					BigInteger id = cyclonView.getNode(BigInteger.valueOf(r.nextInt()));
					if (cyclonView.size() == 0) {
						System.err.println("Empty network");
						return;
					}
					if (peersAddress.get(id).isSource())
					{
						//You should also decrement i
						continue;
					}
					cyclonView.removeNode(id);
					stopAndDestroyPeer(id);
					
				}
			}
		}
	};
	
	//---------------------Edited
	private int snapshotPeerCount;
	private HashMap<PeerAddress, PeerData> snapshotPeers;
	private int dataPointIndex = 0;

	Handler<CollectPeerData> handleCollectPeerData = new Handler<CollectPeerData>() {
		public void handle(CollectPeerData event) {
			if(snapshotPeers != null)
			{
				StateUtil s = new StateUtil(snapshotPeers);
				logger.info("Total Error: " + s.getTotalError().toString());
			}
			snapshotPeers = new HashMap<PeerAddress, PeerData>();

			snapshotPeerCount = 0;
			for (Component peer : peers.values()) {
				trigger(new PeerDataRequest(), peer
						.getPositive(PeerPort.class));
				snapshotPeerCount++;
			}
		}
	};

	Handler<PeerDataResponse> handlePeerDataResponse = new Handler<PeerDataResponse>() {
		public void handle(PeerDataResponse event) {
			PeerAddress node = event.getData().getSelf();
			snapshotPeerCount--;

			if (node != null) {
				snapshotPeers.put(node, event.getData());
			}

			if (snapshotPeerCount == 0) {
				// dump view
				//CyclonDataPoint dataPoint = new CyclonDataPoint(
				//		++dataPointIndex, snapshotPeers);
				//logger.info("DataPoint: \n{}", dataPoint);

				/**
				 * TODO here you can change the code so that instead of printing
				 * out the data points, it collects them for aggregation. E.g.
				 * collect 10 consecutive data points and take their average.
				 */
			}
		}
	};

//-------------------------------------------------------------------	
	Handler<GenerateReport> handleGenerateReport = new Handler<GenerateReport>() {
		public void handle(GenerateReport event) {
			Snapshot.report();
		}
	};

//-------------------------------------------------------------------	
	private final Component createAndStartNewPeer(BigInteger id, int num, boolean isSource) {
		Component peer = create(Peer.class);
		int peerId = ++peerIdSequence;
        InetAddress ip = ipGenerator.generateIP();
		Address address =
//						  new Address(ip, 8058, peerId);
                        new Address(peer0Address.getIp(), peer0Address.getPort(), peerId);

		PeerAddress peerAddress = new PeerAddress(address, id, isSource, num);
		
		connect(network, peer.getNegative(Network.class), new MessageDestinationFilter(address));
		connect(timer, peer.getNegative(Timer.class));
		connect(web, peer.getPositive(Web.class),
				new WebRequestDestinationFilter(peerId));
		
		subscribe(handlePeerDataResponse, peer.getPositive(PeerPort.class));
		
		trigger(new PeerInit(peerAddress, bootstrapConfiguration, cyclonConfiguration, tmanConfiguration, monitorConfiguration), peer.getControl());

		trigger(new Start(), peer.getControl());
		peers.put(id, peer);
		peersAddress.put(id, peerAddress);

		return peer;
	}

//-------------------------------------------------------------------	
	private final void stopAndDestroyPeer(BigInteger id) {
		Component peer = peers.get(id);

		trigger(new Stop(), peer.getControl());
		
		unsubscribe(handlePeerDataResponse, peer.getPositive(PeerPort.class));
		
		disconnect(network, peer.getNegative(Network.class));
		disconnect(timer, peer.getNegative(Timer.class));
		disconnect(web, peer.getPositive(Web.class));
		
		Snapshot.removePeer(peersAddress.get(id));

		peers.remove(id);
		peersAddress.remove(id);

		destroy(peer);
	}

//-------------------------------------------------------------------	
	private final static class MessageDestinationFilter extends
			ChannelFilter<Message, Address> {
		public MessageDestinationFilter(Address address) {
			super(Message.class, address, true);
		}

		public Address getValue(Message event) {
			return event.getDestination();
		}
	}
	
//-------------------------------------------------------------------Edited
	private final static class WebRequestDestinationFilter extends
	ChannelFilter<WebRequest, Integer> {
public WebRequestDestinationFilter(Integer destination) {
	super(WebRequest.class, destination, false);
}

public Integer getValue(WebRequest event) {
	return event.getDestination();
}
}

}

